package com.example.springbootmyelasticjob.config;

import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.JobEventConfiguration;
import com.dangdang.ddframe.job.lite.api.JobScheduler;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import com.example.springbootmyelasticjob.config.dao.JobListDao;
import com.example.springbootmyelasticjob.config.model.JobList;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

@Component
@Slf4j
public class JobRunnerConfig implements CommandLineRunner {
    @Resource
    private ZookeeperRegistryCenter regCenter;

    @Resource
    private JobEventConfiguration jobEventConfiguration;

    @Resource
    private JobListDao jobListDao;

    @Override
    public void run(String... args) throws Exception {
        JobCoreConfiguration coreConfig;
        LiteJobConfiguration liteJobConfiguration = null;

        List<JobList> jobLists = jobListDao.findAll();
        for (JobList jobList : jobLists) {
            coreConfig = JobCoreConfiguration
                    .newBuilder(jobList.getJobName(), jobList.getCron(),
                            Integer.parseInt(jobList.getShardingTotalCount()))
                    .shardingItemParameters(jobList.getShardingItemParameters())
                    .build();
            if ("SIMPLE".equals(jobList.getJobType().trim().toUpperCase())) {
                SimpleJobConfiguration jobConfig=
                        new SimpleJobConfiguration(coreConfig, jobList.getJobClass());
                liteJobConfiguration = LiteJobConfiguration.newBuilder(jobConfig).overwrite(jobList.getOverwrite()==0).build();
            }else if ("DATAFLOW".equals(jobList.getJobType().trim().toUpperCase())){
                DataflowJobConfiguration jobConfig =
                        new DataflowJobConfiguration(coreConfig, jobList.getJobClass(),
                                ("0".equals(jobList.getStreamingProcess().trim())));
                liteJobConfiguration = LiteJobConfiguration.newBuilder(jobConfig).overwrite(jobList.getOverwrite()==0).build();
            }
            new JobScheduler(regCenter,
                    liteJobConfiguration,
                    jobEventConfiguration).init();
        }
    }
}
